

from xpy3lib.XRetryableQuery import XRetryableQuery
from xpy3lib.XRetryableSave import XRetryableSave
from sicost.AbstractDPJob import AbstractDPJob
import sys, datetime, json, time
import pandas as pd
import numpy as np

from numpy import array


class ReductJob(AbstractDPJob):


    def __init__(self,
                 p_df=None, p_downTime=None):
        super(ReductJob, self).__init__()

        self.df = p_df
        self.downTime = p_downTime


        pass


    def execute(self):
        return self.do_execute()


    def do_execute(self):

        super(ReductJob, self).do_execute()

        """
        降频数据Job
        通过传入dataframe，downTime需要降到的频率
        对补全的时间序列进行降低频率
        """

        method = 'nearest'
        # 读取数据集
        new_freq = int(self.downTime * 60)
        # new_freq = 310
        new_freq = str(new_freq)
        print(self.df['datetime'].min())
        # minn = time.mktime(min)
        # print(minn)
        print(self.df['datetime'].max())
        self.df['datetime'] = pd.to_datetime(self.df['datetime'], format="%Y-%m-%d %H:%M:%S")
        if method == 'nearest':
            helper = pd.DataFrame(
                {'datetime': pd.date_range(start=self.df['datetime'].min(), end=self.df['datetime'].max(),
                                           freq=new_freq + 's'), 'remark': 1})
            # helper = pd.DataFrame(helper).set_index('datetime')#将时间列变为索引

            d = pd.merge(self.df, helper, on='datetime', how='outer').sort_values('datetime')
            # d = d.reset_index(drop=True)
            d['datetimeindex'] = d['datetime']
            d.set_index('datetimeindex', inplace=True)
            # d.reindex('datetimeindex', axis='columns')
            d['value'] = d['value'].interpolate(method='nearest',fill_value='extrapolate')
            dd = d[d['remark'] == 1]

            dd.drop(['remark'], axis=1, inplace=True)
            dd = dd.reset_index(drop=True)
            dd['datetime_new'] = dd['datetime'].astype("string")

            dd['timestamp'].fillna(value=0.0, inplace=True)

            def __cal_datetime2timestamp(x):
                if x.timestamp != 0.0:
                    rst = x.timestamp
                else:
                    timeArray = time.strptime(x.datetime_new, "%Y-%m-%d %H:%M:%S")
                    ret = int(time.mktime(timeArray))
                    rst = ret * 1000
                return rst

            dd['timestamp_new'] = dd.apply(lambda x: __cal_datetime2timestamp(x), axis=1)

            dd.drop(['datetime_new'], axis=1, inplace=True)
            dd.drop(['timestamp'], axis=1, inplace=True)
            dd.rename(columns={'timestamp_new': 'timestamp'}, inplace=True)



        else:
            helper2 = pd.DataFrame(
                {'datetime': pd.date_range(start=self.df['datetime'].min(), end=self.df['datetime'].max(),
                                           freq=new_freq + 's'), 'remark': 1})
            helper2['datetime_new'] = helper2['datetime'].astype("string")

            def __cal_datetime2timestamp2(x):

                timeArray = time.strptime(x.datetime_new, "%Y-%m-%d %H:%M:%S")
                ret = int(time.mktime(timeArray))
                rst = ret * 1000
                return rst

            helper2['timestamp2'] = helper2.apply(lambda x: __cal_datetime2timestamp2(x), axis=1)
            helper2.drop(['datetime_new'], axis=1, inplace=True)

            d2 = pd.merge(self.df, helper2, on='datetime', how='outer').sort_values('datetime')
            d2['datetimeindex'] = d2['datetime']
            d2.set_index('datetimeindex', inplace=True)
            # d2.reindex('datetimeindex', axis='columns')
            d2['timestamp2'] = d2['timestamp2'].interpolate(method='nearest')
            # dd2 = d2[d2['remark'] == 1]
            v = ['timestamp2']
            if method == 'mean':
                a = d2.groupby(v)['value'].mean().round(2)
            elif method == 'min':
                a = d2.groupby(v)['value'].min().round(2)
            elif method == 'max':
                a = d2.groupby(v)['value'].max().round(2)

            d2.drop_duplicates(subset=v, keep='first', inplace=True)
            dd = pd.merge(a, d2, on=v, how='left')

        print('finish')
        return dd